package demo_day3;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.HashMap;

public class PartitionProduce {
    // 需求：要生成一个kafka的客户端，完成数据的生产
    public static void main(String[] args) {
        // 1、创建一个map，存放各种配置信息（连接到哪个kafka集群等等）
        // MAP的泛型，不能直接 String String, 因为有的value 是一个class的映射
        HashMap<String,Object> confMap = new HashMap<>();
        // 添加各种配置
        confMap.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"hadoop11:9092,hadoop12:9092,hadoop13:9092");
        // 这里的key 是分区的时候用的
        confMap.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        confMap.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);

        // 2、创建一个producer客户端，接入配置map对象，完成客户端 创建
        KafkaProducer<String,String> producer = new KafkaProducer<>(confMap);

        // 3、基于客户端，完成数据的生产
//        ProducerRecord<String,String> record = new ProducerRecord<>("topicB","from_java");
//        producer.send(record); // 发送数据
        // 需求一：发送10条数据到kafka，在消费的时候数据是有序的
        // 需求二：有很多商品需要进行秒杀，那么就要求每一个商品的下单信息是有序的
        for (int i = 0; i < 10; i++) {
            // 一、指明分区，固定每次的发送都发送到哪个分区中
            producer.send(new ProducerRecord<>("topicB",0,null,"hello"+i));
            // 二、指明key的值，kafka会根据key的值进行分区的识别，完成发送
            producer.send(new ProducerRecord<>("topicB","2","hello kafka" + i));
        }


        // 4、关闭资源
        producer.close();
    }
}
